package com.sxzjrj

import com.alibaba.fastjson.{JSON, JSONObject}
import com.sxzjrj.constant.Constant
import com.sxzjrj.utils.{CmccUtils, JdbcUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext, TaskContext}

/**
  * Created by ljj in 2019/3/21
  *
  */
object CmccSparkCore {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("充值监控平台")

    val sc = new SparkContext(conf)

    //    val result = sc.textFile("hdfs:/data/cmcc.log")
    val result = sc.textFile("e:/cmcc.log")
      .filter(_.contains(Constant.RECHARGENOTIFYREQ.toString))
      .map(line => {

        var province: String = "000"
        var money: Int = 0
        var time: Long = 0L
        var succOrder: Int = 0

        try {

          val bean: JSONObject = JSON.parseObject(line)
          val bussinessRst = bean.getString("bussinessRst")
          val flag = bussinessRst.equals("0000")

          val costTime = CmccUtils.getCostTime(bean.getString("requestId"), bean.getString("receiveNotifyTime"))

          province = if (bean.getString("provinceCode").length > 0) bean.getString("provinceCode") else "000"

          succOrder = if (flag) 1 else 0

          time = if (flag) costTime else 0L

          money = if (flag) {
            if (bean.getString("chargefee").length > 0) bean.getString("chargefee").toInt else 0
          } else 0
        } catch {
          case e: Exception => e.printStackTrace()
        }
        (province, (1, succOrder, money, if (time >= 0L) time else time * (-1)))
      }).cache()
      .reduceByKey((a, b) => (a._1 + b._1, a._2 + b._2, a._3 + b._3, a._4 + b._4))
    //      .sortBy(-_._2._3)
    //      .foreachPartition(iter => {
    //        iter.foreach(tp => println(TaskContext.getPartitionId(), tp))
    //      })
    //      .foreachPartition(iter => {
    //        iter.foreach(tp => println(TaskContext.getPartitionId(), tp))
    //      })
    //     result.saveAsTextFile("hdfs:/cmccTest" )

result.saveAsTextFile("e:\\cmccTest")

//    JdbcUtils.saveData2Mysql(result)


    sc.stop()

  }

}
